﻿using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Reactive;
using Microsoft.WindowsAzure.ServiceRuntime;
using System.Diagnostics.Contracts;
using System.ServiceModel;
using System.Threading.Tasks;
using System.Collections.Concurrent;

namespace RxCloud
{
    /// <summary>
    /// Represents queue notificationsClient client.
    /// </summary>
    public sealed class NotificationsClient : IDisposable
    {
        class NotificationsObserver : IObserver<Unit>
        {
            private readonly List<ChannelFactory<IQueueNotifier>> channelFactories;
            private readonly string queueName;

            private Random rnd = new Random();
            private List<IObserver<Unit>> notifiers = new List<IObserver<Unit>>();

            public NotificationsObserver(List<ChannelFactory<IQueueNotifier>> channelFactories, string queueName)
            {
                Contract.Requires(channelFactories != null);
                Contract.Requires(queueName != null);
                Contract.Ensures(this.channelFactories != null);
                Contract.Ensures(this.queueName != null);

                this.channelFactories = channelFactories;
                this.queueName = queueName;


            }

            public bool IsActive { get; private set; }

            public void Start()
            {
                if (!this.IsActive)
                {
                    BuildObservers();
                    this.IsActive = true;
                }
            }

            public void Stop()
            {
                if (this.IsActive)
                {
                    notifiers.Clear();
                    this.IsActive = false;
                }
            }

            private void BuildObservers()
            {
                Contract.Ensures(Contract.ForAll(notifiers, e => e != null));
                notifiers.AddRange(this.channelFactories
                                       .Select(cf => NotificationsClient.GetQueueNotifierObserver(cf.CreateChannel(), this.queueName)));
            }

            #region IObserver members
            public void OnCompleted()
            {
                if (this.IsActive && notifiers.Count > 0)
                {
                    //Send completion to all workers
                    notifiers.ForEach(obs => obs.OnCompleted());
                }
            }

            public void OnError(Exception error)
            {
                if (this.IsActive && notifiers.Count > 0)
                {
                    notifiers.ForEach(obs => obs.OnError(error));
                }
            }

            public void OnNext(Unit value)
            {
                if (this.IsActive && notifiers.Count > 0)
                {
                    //Send notification to random worker
                    notifiers[rnd.Next(notifiers.Count)].OnNext(value);
                }
            }
            #endregion
        }

        private static readonly object syncRoot = new object();

        private readonly Role role;
        private readonly string endpointName;
        private readonly string servicePath;

        private List<ChannelFactory<IQueueNotifier>> channelFactories = new List<ChannelFactory<IQueueNotifier>>();
        private bool isActive = false;
        private ConcurrentDictionary<string, NotificationsObserver> observers = new ConcurrentDictionary<string, NotificationsObserver>();

        #region Constructors
        /// <summary>
        /// Initializes queue notificationsClient client
        /// </summary>
        /// <param name="role">Role hosts notificationsClient server</param>
        /// <param name="endpointName">Internal communication endpoint name</param>
        public NotificationsClient(Role role, string endpointName)
            : this(role, endpointName, Constants.DefaultEndpointName)
        { }

        /// <summary>
        /// Initializes queue notificationsClient client
        /// </summary>
        /// <param name="role">Role hosts notificationsClient server</param>
        public NotificationsClient(Role role)
            : this(role, Constants.DefaultEndpointName)
        { }

        /// <summary>
        /// Initializes queue notificationsClient client
        /// </summary>
        /// <param name="role">Role hosts notificationsClient server</param>
        /// <param name="endpointName">Internal communication endpoint name</param>
        /// <param name="servicePath">Relative service uri</param>
        public NotificationsClient(Role role, string endpointName, string servicePath)
        {
            Contract.Requires(role != null);
            Contract.Requires(endpointName != null);
            Contract.Requires(servicePath != null);
            Contract.Ensures(this.role != null);
            Contract.Ensures(this.endpointName != null);
            Contract.Ensures(this.servicePath != null);

            this.role = role;
            this.endpointName = endpointName;
            this.servicePath = servicePath;
        }
        #endregion

        /// <summary>
        /// Returns IObserver to send notificationsClient.
        /// </summary>
        /// <param name="queueName">Queue name.</param>
        /// <returns>Observer.</returns>
        public IObserver<Unit> GetQueueNotifierObserver(string queueName)
        {
            Contract.Requires(queueName != null);
            Contract.Ensures(Contract.Result<IObserver<Unit>>() != null);
            Start();

            var obs = GetObserver(queueName);
            obs.Start();
            return obs;
        }

        /// <summary>
        /// Restarts client. Useful on configuration change.
        /// </summary>
        public void Restart()
        {
            Stop();
            Start();
        }

        private NotificationsObserver GetObserver(string queueName)
        {
            Contract.Requires(queueName != null);
            Contract.Ensures(Contract.Result<IObserver<Unit>>() != null);

            return observers.GetOrAdd(queueName, new NotificationsObserver(this.channelFactories, queueName));
        }

        private void Start()
        {
            if (!this.isActive)
            {
                lock (syncRoot)
                {
                    if (!this.isActive)
                    {
                        CreateChannels();
                        StartObservers();
                        this.isActive = true;
                    }
                }
            }
        }

        private void Stop()
        {
            if (this.isActive)
            {
                lock (syncRoot)
                {
                    if (this.isActive)
                    {
                        StopObservers();
                        CloseChannels();
                        this.isActive = false;
                    }
                }
            }
        }

        private void CreateChannels()
        {
            channelFactories.AddRange(role.Instances.Select(CreateChannel));

            //Open all channelFactories
            Task.WaitAll(channelFactories.Select(cf => Task.Factory.FromAsync(cf.BeginOpen, cf.EndOpen, null)).ToArray());
        }

        private ChannelFactory<IQueueNotifier> CreateChannel(RoleInstance instance)
        {
            Contract.Requires(instance != null);
            Contract.Ensures(Contract.Result<ChannelFactory<IQueueNotifier>>() != null);

            var endpoint = instance.InstanceEndpoints[this.endpointName];
            if (endpoint == null)
            {
                throw new InvalidOperationException(string.Format("No endpoint with name \"{0}\"", this.endpointName));
            }

            var endpointAddress = string.Format(Constants.ServiceEndpointAddress, endpoint.IPEndpoint, servicePath, Utils.GetScheme(endpoint.Protocol));
            return new ChannelFactory<IQueueNotifier>(Utils.GetBinding(endpoint.Protocol), endpointAddress);
        }

        private void CloseChannels()
        {
            //Close all channelFactories
            Task.WaitAll(channelFactories.Select(cf => Task.Factory.FromAsync(cf.BeginClose, cf.EndClose, null)).ToArray());

            channelFactories.Clear();
        }

        private void StartObservers()
        {
            foreach (var item in this.observers)
            {
                item.Value.Start();
            }
        }

        private void StopObservers()
        {
            foreach (var item in this.observers)
            {
                item.Value.Stop();
            }
        }

        private static IObserver<Unit> GetQueueNotifierObserver(IQueueNotifier proxy, string queueName)
        {
            Contract.Requires(proxy != null);
            Contract.Requires(queueName != null);
            Contract.Ensures(Contract.Result<IObserver<Unit>>() != null);

            return Observer.Create<Unit>(
                        _ => proxy.MessageAdded(queueName),
                        _ => proxy.NoMoreMessages(queueName),
                        () => proxy.NoMoreMessages(queueName)
                    );
        }

        /// <summary>
        /// Closes all notificaitions channels.
        /// </summary>
        public void Dispose()
        {
            Stop();
        }
    }
}
